package com.wtwd.wtwddevice.deviceserver.rocketMq;

import com.wtwd.device.mqtentity.MessageBase;
import com.wtwd.device.properties.MqProperties;
import com.wtwd.wtwddevice.deviceserver.handler.MsgHandlerFactory;
import io.netty.channel.DefaultChannelId;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public class MqReceiveMsg {

    private MqProperties mqProperties;

    public MqReceiveMsg(MqProperties mQproperties) {
        this.mqProperties = mQproperties;
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mQproperties.getDeviceServiceConsumeGroup());
        try {
            consumer.setNamesrvAddr(mQproperties.getNamesrvAddr());
            consumer.subscribe(mQproperties.getDeviceConsumeTopic(), mQproperties.getDeviceMsgTag());
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt messageBaseExt : msgs) {
                        MessageBase messageBase = MessageBase.deserialize(messageBaseExt.getBody());
                        log.info("topic:{},tag:{}  mq 接收消息 msg: {}",mQproperties.getDeviceConsumeTopic(),mQproperties.getDeviceMsgTag(),messageBase.toString());
                        try {
                            MsgHandlerFactory.getMsgHandler().sendMessage(messageBase.getWtProtocol());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } catch (Exception e) {
                            int reconsumeTimes = messageBaseExt.getReconsumeTimes();
                            log.error("处理消息异常 msg {},{},{}", messageBaseExt.getReconsumeTimes(), messageBase,e.getMessage());
                            if (reconsumeTimes >= mQproperties.getConsumeRetryTimes()) {
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            DefaultChannelId.newInstance();
            consumer.start();
            log.info("成功订阅消费者组:{} 主题: {} 消费者tag;{}",mqProperties.getDeviceServiceConsumeGroup(),mQproperties.getDeviceConsumeTopic(),mQproperties.getDeviceMsgTag());
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }
}
